package org.example
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions.{col, desc}
import org.apache.spark.sql.types.{DataTypes, IntegerType, StringType, StructField, StructType}
object spark_SQL3 {
  def main(args: Array[String]): Unit = {
    //  创建spark运行环境
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
    //    设置表的结构（列名）和数据类型
    val schemaUser = StructType(Seq(
      StructField("id", IntegerType),
      StructField("gender", StringType),
      StructField("age", IntegerType),
      StructField("occupation", IntegerType),
      StructField("localtion",StringType),
    ))
    val user = spark.read.option("sep","::").schema(schemaUser)
      .csv("src/main/resources/movies/users.dat")
    user.show(5)
//    1.查询 替换 where/filter  udf
    user.where("gender = 'F' and age = 18").show(3)
    user.filter("gender = 'F' and age = 18").show(3)

    spark.udf.register("replace", (x:String) => {
      x match {
        case "M" => 0
        case "F" => 1
      }
    })
    user.selectExpr("id","replace(gender) as sexual","age").show(3)
//    user.select("id","age","location").show(3)
//    2.排序
    user.orderBy(desc("age")).show(5)
    user.sort(-user("id")).show(5)
//    3.分组
    user.groupBy("gender").count().show()
//    4.连接 join(DataFrame ,"列名"，"连接方式left_outer")
//    5.练习:求18岁女生评分电影为5分的所有电影名
    val schemaRatings = StructType(Seq(
      StructField("id", IntegerType),
      StructField("moviesname", StringType),
      StructField("score", IntegerType),
      StructField("occupation", IntegerType),
))
    val ratings = spark.read.option("sep", "::").schema(schemaRatings)
      .csv("src/main/resources/movies/ratings.dat")
    val schemaMovies = StructType(Seq(
      StructField("MovieID", IntegerType),
      StructField("Title", StringType),
      StructField("Genres", StringType)
    ))
    val movies = spark.read.option("sep", "::").schema(schemaMovies)
      .csv("src/main/resources/movies/movies.dat")
    val female8Users = user.filter(col("age") === 18 && col("gender") === "F")
      .select("id")
    val result = ratings
      .join(female8Users,"id")
      .filter(col("rating") === 5)
      .join(movies,"MovieID")
      .select("Title").distinct()
    result.show(truncate = false)


    sc.stop()
  }
}
